package org.gthelper.hdpWritable;

  
//import java.awt.image.RenderedImage;
//import java.awt.image.RenderedImage;
import java.io.DataInput;  
import java.io.DataOutput;  
import java.io.IOException;  
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
//import java.util.Arrays;





import org.apache.hadoop.io.Text;  
import org.apache.hadoop.io.WritableComparable;  
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;

import javax.media.jai.remote.SerializableRenderedImage;



/**
 * HdpRenderedImageWritable
 * @author PIG
 * 自定义hadoop的WritableComparable类型，实现将WritableRaster作为mapreduce的value
 */
@SuppressWarnings("rawtypes")
public class HdpRenderedImageWritable implements WritableComparable
{
	public Text name;
	public int number_x;//图片列号
	public int number_y;//图片行号
	public int position_x;//图片左上角位置
	public int position_y;
	
	public SerializableRenderedImage sri;
	
/**
 * 必须有的默认构造器	
 */
public HdpRenderedImageWritable()
{
	this.name=new Text();
	this.number_x = 0;
	this.number_y = 0;
	this.position_x = 0;
	this.position_y = 0;
	this.sri=null;
	
	}




public void set(Text name,int numberx,int numbery,int posx,int posy,SerializableRenderedImage sri)
{
	this.name=name;
	this.number_x = numberx;
	this.number_y = numbery;
	this.position_x = posx;
	this.position_y = posy;
	this.sri=sri;
}




	/**
	 * readFields
	 * @see org.apache.hadoop.io.Writable#readFields(java.io.DataInput)
	 * writableComparable接口方法之一，用于反序列化
	 */
	@Override
	public void readFields(DataInput arg0) throws IOException{
		// TODO 自动生成的方法存根
		name.readFields(arg0);
		this.number_x = arg0.readInt();
		this.number_y = arg0.readInt();
		this.position_x = arg0.readInt();
		this.position_y = arg0.readInt();
		//long length=arg0.readLong();
		
		//通过byte[]反序列化WirtableRaster
		int length = (int)arg0.readLong();
		byte[] bytes = new byte[length];
		arg0.readFully(bytes, 0, length);
		ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
		ObjectInputStream ois = new ObjectInputStream(bais);
		try {
			this.sri=(SerializableRenderedImage)ois.readObject();
		} catch (ClassNotFoundException e) {
			// TODO 自动生成的 catch 块
			e.printStackTrace();
		}
		
	}

	
	
	/**
	 * write
	 * @see org.apache.hadoop.io.Writable#write(java.io.DataOutput)
	 * writableComparable接口方法之一，用于序列化
	 */
	@Override
	public void write(DataOutput arg0) throws IOException {
		// TODO 自动生成的方法存根
		name.write(arg0);
		arg0.writeInt(this.number_x);
		arg0.writeInt(this.number_y);
		arg0.writeInt(this.position_x);
		arg0.writeInt(this.position_y);
		
		//序列化为byte[]再传输
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		ObjectOutputStream oos = new ObjectOutputStream(baos);
        oos.writeObject(sri);
        byte[] bytes = baos.toByteArray();
        long length=bytes.length;
        arg0.writeLong(length);
        arg0.write(bytes);
	}

	
	//比较器
	@Override
	public int compareTo(Object arg0) {
		// TODO 自动生成的方法存根
		return 0;
	}
	 
	
	/**
	 * RwComparator
	 * @author PIG
	 * 实现HdpRenderedImageWritable的默认比较器
	 */
	public static class RwComparator extends WritableComparator
	{
		private static final Text.Comparator TEXT_COMPARATOR= new Text.Comparator();
		
		protected RwComparator() {  
	        super(HdpRenderedImageWritable.class);  
	    }  
		
		@Override  
	    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {  
	        try {  
	              
	            int nameL1= WritableUtils.decodeVIntSize(b1[s1])+readVInt(b1,s1);  
	            int nameL2=WritableUtils.decodeVIntSize(b2[s2])+readVInt(b2,s2);  
	            //先比较name  
	            int cmp = TEXT_COMPARATOR.compare(b1,s1,nameL1,b2,s2,nameL2);  
	            if(cmp!=0){  
	                return cmp;  
	            }  
	            //再比较role  
	            return TEXT_COMPARATOR.compare(b1,s1+nameL1,l1-nameL1,b2,s2+nameL2,l2-nameL2);  
	        } catch (IOException e) {  
	            throw new IllegalArgumentException();  
	        }  
	    }  
	}
	    static {  
	        //注册comprator,这样MapReduce使用HdpRenderedImageWritable时就会直接调用RwComparator  
	        WritableComparator.define(HdpRenderedImageWritable.class,new RwComparator());  
	    }  
		
	
	
}
